package com.example.mq.mqserver;

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.ConsumerManager;
import com.example.mq.mqserver.datacenter.DiskDataCenter;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 通过这个类 表示 虚拟主机
 * 每个虚拟主机下面 管理着自己的交换机，队列，绑定，消息数据
 * 同时提供 API 供上层调用
 * 这个类是对业务的逻辑整合，就需要处理异常了
 */
public class VirtualHost {
    private String virtualName;
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    // 交换机锁对象
    private final Object exchangeLocker = new Object();
    // 队列锁对象
    private final Object queueLocker = new Object();

    public VirtualHost(String virtualName) {
        this.virtualName = virtualName;
        // 对于 MemoryDataCenter，不需要额外的初始化操作，只要把对象 New 出来就行了
        // 但是针对 DiskDataCenter 来说，则需要进行初始化操作，建库建表和初始化数据的设定
        diskDataCenter.init();
        // 另外还需要将 硬盘数据，进行恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | ClassNotFoundException | MqException e) {
            e.printStackTrace();
            System.out.println("[Virtual] 恢复数据到内存失败！");
        }
    }

    public String getVirtualName() {
        return virtualName;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }


    // 创建交换机
    // 如果交换机不存在，就创建， 存在则直接返回
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) {
        // 把交换机的名字，加上虚拟主机作为前缀
        exchangeName = virtualName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1. 判定该交换机是否存在，直接通过内存查询
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    // 该交换机已经存在！
                    System.out.println("[VirtualHost] 交换机已经存在！exchangeName=" + exchangeName);
                    return true;
                }
                // 2. 真正创建交换机，先构造 Exchange 对象
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setDurable(durable);
                exchange.setType(exchangeType);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                // 3. 把交换机对象写入硬盘, durable为true时才写入
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                // 4. 把交换机写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成！exchangeName=" + exchangeName);
                // 上述逻辑，先写硬盘，再写内存。
                // 因为硬盘更容易写失败，一旦失败，就不写内存了
                // 要是先写内存，内存写成功了，硬盘写失败了，还需要把内存数据清理了，就比较麻烦
            }
            return true;
        } catch (Exception e) {
            System.out.println(("[VirtualHost] 交换机创建失败！exchangeName=" + exchangeName));
            e.printStackTrace();
            return false;
        }

    }

    // 删除交换机
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1. 先找到对应的交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MqException("[VirtualHost] 交换机不存在！无法删除！exchangeName=" + exchangeName);
                }
                // 2. 删除硬盘数据
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                // 3. 从内存中删除
                memoryDataCenter.deletaExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功！exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println(("[VirtualHost] 交换机删除失败！exchangeName=" + exchangeName));
            e.printStackTrace();
            return false;
        }
    }

    // 创建队列
    public boolean queueDelcare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments) {
        // 把队列的名字，拼接上虚拟主机名字
        queueName = virtualName + queueName;
        try {
            synchronized (queueLocker) {
                // 1. 判定队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在！queueName=" + queueName);
                    return true;
                }
                // 2. 不存在则创建队列
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                // 3. 将队列写入到硬盘durable为true时才写入
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                // 4. 将队列写入到内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功！queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println(("[VirtualHost] 队列创建失败！queueName=" + queueName));
            e.printStackTrace();
            return false;
        }
    }

    // 队列删除
    public boolean queueDelete(String queueName) {
        queueName = virtualName + queueName;
        try {
            synchronized (queueLocker) {
                // 1. 查询队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue == null) {
                    throw new MqException("[VirtualHost] 队列不存在！无法删除！");
                }
                // 2. 存在进行，先在硬盘删除
                if (existsQueue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                // 3. 在内存中删除
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除成功！queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println(("[VirtualHost] 队列删除失败！queueName=" + queueName));
            e.printStackTrace();
            return false;
        }
    }

    // 创建绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        queueName = virtualName + queueName;
        exchangeName = virtualName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 判断当前的绑定是否已经存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null) {
                        throw new MqException("[VirtualHost] binding已经存在！exchangeName=" + exchangeName + ", queueName=" + queueName);
                    }
                    // 2. 验证 bindingKey 是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey非法！bindingKey=" + bindingKey);
                    }
                    // 3. 不存在就创建绑定
                    Binding binding = new Binding();
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    binding.setBindingKey(bindingKey);
                    // 4. 获取下绑定 对应的 队列和交换机是否存在
                    Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                    MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                    if (existsExchange == null) {
                        throw new MqException("[VirtualHost] 交换机不存在！exchangeName=" + exchangeName);
                    }
                    if (existsQueue == null) {
                        throw new MqException("[VirtualHost] 队列不存在！queueName=" + queueName);
                    }
                    // 5. 先写入硬盘，需要判断当前 交换机和队列是否都持久化
                    if (existsQueue.isDurable() && existsExchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    // 6. 再写入内存
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost]  绑定创建成功！ exchangeName=" + exchangeName + ", queueName=" + queueName);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost]  绑定创建失败！ exchangeName=" + exchangeName + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除绑定
    public boolean queueUnbind(String exchangeName, String queueName) {
        exchangeName = virtualName + exchangeName;
        queueName = virtualName + queueName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 获取绑定是否存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding == null) {
                        throw new MqException("[VirtualHost] 绑定不存在！无法删除！exchangeName=" + exchangeName +
                                ", queueName=" + queueName);
                    }
//            // 2. 获取 对应的队列和交换机
//            Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
//            MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
//            if (existsExchange == null) {
//                throw new MqException("[VirtualHost] 交换机不存在！exchangeName=" + exchangeName);
//            }
//            if (existsQueue == null) {
//                throw new MqException("[VirtualHost] 队列不存在！queueName=" + queueName);
//            }
                    // 3. 删除硬盘上的数据 需要判断当前 交换机和队列都是持久化
                    diskDataCenter.deleteBinding(existsBinding);

                    // 4. 从内存中删除绑定
                    memoryDataCenter.deleteBinding(existsBinding);
                    System.out.println("[VirtualHost] 删除绑定成功！");
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除绑定失败！");
            return false;
        }
    }

    // 发送消息到指定的交换机/队列中
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        try {
            // 1. 转换交换机名字
            exchangeName = virtualName + exchangeName;
            // 2. 检查 routingKey 是否合法
            if (!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey 非法! routingKey=" + routingKey);
            }
            // 3. 查找交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
            }
            // 4. 判断交换机类型
            if (exchange.getType() == ExchangeType.DIRECT) {
                // 按照直接交换机的方式来转发消息
                // 以 routingKey 为队列名，直接将 消息写入 队列中
                // 此时可以无视绑定消息
                String queueName = virtualName + routingKey;
                // 5. 构造消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                // 6. 查找该队列名所对应的 队列是否存在
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在！queueName=" + queueName);
                }
                // 7. 队列存在，则直接写入消息
                sendMessage(queue, message);

            } else {
                // 按照 FANOUT 和 TOPIC 方式
                // 5. 找到该交换机关联的所有绑定消息，并遍历这些绑定消息
                ConcurrentHashMap<String, Binding> bindings = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindings.entrySet()) {
                    // 1) 获取绑定对象，判断该队列是否存在
                    Binding binding = entry.getValue();
                    // 2) 查看当前绑定里面的队列名，能不能查到相应队列 (此处不设计转发规则)
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        System.out.println("[VirtualHost] basicPublish 发送消息时，发现队列不存在！queueName=" + binding.getQueueName());
                        continue;
                    }
                    // 2) 构造消息对象
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // 3) 判断这个消息能否转发给这个队列
                    // 如果是 fanout, 所有绑定的队列都要转发的
                    // 如果是 topic, 还需要 判断 routingKey 和 BindingKey 是否匹配
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    // 4) 真正转发消息给队列
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        // 1. 此处发送消息，就是把消息 写入到 硬盘 和 内存
        int deliverMode = message.getDeliverMode();
        // deliverMode 为2持久化  为1不持久化
        if (deliverMode == 2) {
            diskDataCenter.sendMessage(queue, message);
        }
        // 写入内存
        memoryDataCenter.sendMessage(queue, message);

        //  通知消费者可以消费消息了
        consumerManager.notifyConsume(queue.getName());
    }

    // 订阅消息
    // 添加一个队列的订阅者，当队列收到消息后，就要把消息推给对应的订阅者
    // consumeTage: 消费者身份标识
    // autoAck: 消息被消费完成后，应答的方式，为 true 自动应答，为 false 手动应答
    // consumer: 是一个回调函数，此处类型设置成函数接口，这样后续调用 basicConsume 并且传实参的时候，就可写作 lambda 表达式
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        // 构造一个 consumerEnv 对象，把这个对应的队列找到，再把这个 Consumer 对象添加到该队列中
        queueName = virtualName + queueName;
        System.out.println("queueName=" + queueName);
        try {
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume 成功 queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicConsume 失败 queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 确认消息
    public boolean basicAck(String queueName, String messageId) {
        try {
            // 1. 获取到消息和队列
            queueName = virtualName +queueName;
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MqException("[VirtualHost] 要确认的消息不存在！messageId=" + messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MqException("[VirtualHost] 要确认的队列不存在！messageId=" + messageId);
            }
            // 2. 删除硬盘上数据
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            // 3. 删除内存中心数据
            memoryDataCenter.removeMessage(messageId);
            // 4. 删除待确认集合中的数据
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] basicAck成功！消息被成功确认！queueName" + queueName +
                    ", messageId=" + messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicAck失败！消息确认失败！queueName=" + queueName +
                    ", messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }

}
